#例【6-1】
import json
import uuid
import time
from datetime import datetime

# 定义基础的Prompt对象，用于存储每一轮输入与输出
class Prompt:
    def __init__(self, role: str, content: str, parent_id: str = None, metadata: dict = None):
        # 生成唯一标识
        self.id = str(uuid.uuid4())
        self.role = role                # user, assistant, system, tool等角色
        self.content = content          # 具体的提示文本内容
        self.parent_id = parent_id      # 上一级Prompt的ID，用于构建嵌套上下文链
        self.metadata = metadata or {}  # 附加元数据，如时间戳、状态等
        self.timestamp = datetime.utcnow().isoformat()

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "role": self.role,
            "content": self.content,
            "parent_id": self.parent_id,
            "metadata": self.metadata,
            "timestamp": self.timestamp
        }

# 定义ContextChain，用于管理多轮、多级Prompt链的嵌套结构
class ContextChain:
    def __init__(self):
        # 使用字典存储所有Prompt，键为Prompt的ID
        self.prompts = {}

    def add_prompt(self, prompt: Prompt):
        # 添加新Prompt对象到上下文链中
        self.prompts[prompt.id] = prompt

    def get_chain(self, end_id: str) -> list:
        """
        根据终止Prompt的ID，追溯整个上下文链，
        返回从根节点到当前节点的Prompt列表
        """
        chain = []
        current = self.prompts.get(end_id)
        while current:
            chain.insert(0, current)
            if current.parent_id:
                current = self.prompts.get(current.parent_id)
            else:
                break
        return chain

    def print_chain(self, end_id: str):
        # 打印从根节点到指定Prompt的上下文链
        chain = self.get_chain(end_id)
        print("=== 当前上下文链 ===")
        for prompt in chain:
            print(f"[{prompt.role}] {prompt.content} ({prompt.id[:6]})")
        print("=====================")

# 定义Task类，用于表示一个任务（可以为主任务或子任务）
class Task:
    def __init__(self, task_name: str, description: str, parent_task=None):
        """
        初始化任务对象，可包含子任务
        :param task_name: 任务名称
        :param description: 任务描述
        :param parent_task: 父任务对象
        """
        self.task_id = str(uuid.uuid4())
        self.task_name = task_name
        self.description = description
        self.parent_task = parent_task
        self.subtasks = []       # 子任务列表
        self.context_chain = ContextChain()  # 每个任务维护自己的上下文链

    def add_subtask(self, subtask):
        # 添加子任务，并设置该子任务的父任务为当前任务
        subtask.parent_task = self
        self.subtasks.append(subtask)

    def add_prompt(self, prompt: Prompt):
        # 将提示词添加到任务的上下文链中
        self.context_chain.add_prompt(prompt)

    def get_context(self):
        # 返回整个任务的上下文链
        return self.context_chain.prompts

    def print_task_context(self, prompt_id: str):
        # 打印当前任务中的上下文链信息
        self.context_chain.print_chain(prompt_id)

# 定义一个用于状态转移中的语义保持机制的函数
def preserve_semantics(chain: ContextChain, current_prompt: Prompt) -> str:
    """
    对当前上下文链中的所有Prompt进行合并，
    保持上下文语义的连续性与完整性。返回合并后的文本。
    :param chain: 当前任务的上下文链管理器
    :param current_prompt: 当前最新的Prompt对象
    :return: 合并后的上下文文本字符串
    """
    prompts = chain.get_chain(current_prompt.id)
    # 通过换行符连接每个Prompt的内容，确保顺序一致
    merged_text = "\n".join([p.content for p in prompts])
    return merged_text

# 主任务场景：简历筛选智能体
def main():
    # 创建主任务：评估候选人是否符合岗位要求
    main_task = Task("简历筛选", "评估候选人是否符合岗位要求，根据简历及历史对话判断综合匹配度")

    # 添加系统初始化提示，作为上下文链的根节点
    system_prompt = Prompt(
        role="system",
        content="启动候选人评估任务，确定岗位要求为：熟悉Python、经验丰富的后端开发。"
    )
    main_task.add_prompt(system_prompt)

    # 子任务1：解析简历内容
    subtask_parse = Task("解析简历", "提取候选人简历中的基本信息，如教育背景、工作经验和技能")
    main_task.add_subtask(subtask_parse)

    # 添加用户输入的简历提示，作为子任务1的上下文输入
    resume_prompt = Prompt(
        role="user",
        content="候选人简历：张三，计算机科学本科，5年后端开发经验，精通Python和数据库设计。",
        parent_id=system_prompt.id
    )
    subtask_parse.add_prompt(resume_prompt)

    # 子任务2：匹配岗位需求
    subtask_match = Task("匹配岗位", "将候选人的技能与岗位要求进行匹配，计算匹配度。")
    main_task.add_subtask(subtask_match)

    # 添加工具调用提示，引用子任务1的输出作为输入
    match_prompt = Prompt(
        role="assistant",
        content="根据简历内容判断候选人张三的技能匹配度，结果为：",
        parent_id=resume_prompt.id
    )
    subtask_match.add_prompt(match_prompt)

    # 子任务3：生成综合评价摘要
    subtask_summary = Task("生成摘要", "综合各子任务结果生成候选人评价摘要。")
    main_task.add_subtask(subtask_summary)

    # 添加用户输入提示，要求生成摘要，需要引用子任务2的内容
    summary_prompt = Prompt(
        role="user",
        content="请综合以上信息，生成详细的候选人综合评价摘要。",
        parent_id=match_prompt.id
    )
    subtask_summary.add_prompt(summary_prompt)

    # 上下文转移与语义保持：在主任务中将所有子任务结果合并保持语义一致性
    # 对子任务2的最后一条Prompt执行上下文合并
    merged_context = preserve_semantics(subtask_match.context_chain, match_prompt)
    # 将合并后的上下文作为状态标签注入到子任务3中
    transferred_prompt = Prompt(
        role="assistant",
        content=f"整合匹配结果：{merged_context}",
        parent_id=summary_prompt.id
    )
    subtask_summary.add_prompt(transferred_prompt)

    # 输出各任务中的上下文信息，验证嵌套与转移结构
    print("=== 主任务上下文链 ===")
    main_task.context_chain.print_chain(system_prompt.id)
    print("\n=== 子任务1（解析简历）上下文链 ===")
    subtask_parse.context_chain.print_chain(resume_prompt.id)
    print("\n=== 子任务2（匹配岗位）上下文链 ===")
    subtask_match.context_chain.print_chain(match_prompt.id)
    print("\n=== 子任务3（生成摘要）上下文链 ===")
    subtask_summary.context_chain.print_chain(transferred_prompt.id)

    # 最终，通过合并所有子任务上下文，生成完整的任务描述文本
    final_context = (
        preserve_semantics(main_task.context_chain, system_prompt) + "\n\n" +
        preserve_semantics(subtask_parse.context_chain, resume_prompt) + "\n\n" +
        preserve_semantics(subtask_match.context_chain, match_prompt) + "\n\n" +
        preserve_semantics(subtask_summary.context_chain, transferred_prompt)
    )

    # 输出最终合并后的上下文文本
    print("\n=== 最终合并的上下文文本 ===")
    print(final_context)

if __name__ == "__main__":
    main()



#例【6-2】
import time
from typing import List, Dict, Optional

class ContextManager:
    """
    上下文管理器，负责动态管理多任务对话中的上下文信息。
    """

    def __init__(self, max_length: int = 5):
        """
        初始化上下文管理器。

        :param max_length: 上下文存储的最大长度，超过该长度将进行裁剪。
        """
        self.contexts: Dict[str, List[Dict[str, str]]] = {}
        self.max_length = max_length

    def add_context(self, task_id: str, role: str, content: str):
        """
        添加上下文信息。

        :param task_id: 任务ID，用于区分不同的任务上下文。
        :param role: 角色名称，如'user'或'assistant'。
        :param content: 上下文内容。
        """
        if task_id not in self.contexts:
            self.contexts[task_id] = []
        self.contexts[task_id].append({'role': role, 'content': content, 'timestamp': time.time()})
        # 如果超过最大长度，移除最早的上下文
        if len(self.contexts[task_id]) > self.max_length:
            self.contexts[task_id].pop(0)

    def get_context(self, task_id: str) -> List[Dict[str, str]]:
        """
        获取指定任务的上下文信息。

        :param task_id: 任务ID。
        :return: 上下文信息列表。
        """
        return self.contexts.get(task_id, [])

    def remove_context(self, task_id: str):
        """
        移除指定任务的上下文信息。

        :param task_id: 任务ID。
        """
        if task_id in self.contexts:
            del self.contexts[task_id]

    def clear_all_contexts(self):
        """
        清除所有任务的上下文信息。
        """
        self.contexts.clear()

class TaskHandler:
    """
    任务处理器，模拟处理不同的任务，并利用上下文管理器管理上下文。
    """

    def __init__(self, context_manager: ContextManager):
        """
        初始化任务处理器。

        :param context_manager: 上下文管理器实例。
        """
        self.context_manager = context_manager

    def handle_task(self, task_id: str, user_input: str):
        """
        处理用户输入的任务。

        :param task_id: 任务ID。
        :param user_input: 用户输入的内容。
        """
        # 添加用户输入到上下文
        self.context_manager.add_context(task_id, 'user', user_input)

        # 根据用户输入决定任务类型
        if '天气' in user_input:
            response = self.get_weather_info(user_input)
        elif '新闻' in user_input:
            response = self.get_news_info(user_input)
        else:
            response = "抱歉，我无法理解您的请求。"

        # 添加助手的响应到上下文
        self.context_manager.add_context(task_id, 'assistant', response)

        # 输出当前上下文
        self.display_context(task_id)

    def get_weather_info(self, location: str) -> str:
        """
        获取天气信息的模拟方法。

        :param location: 地点信息。
        :return: 天气信息字符串。
        """
        # 模拟天气信息
        return f"{location}的天气是晴天，气温25摄氏度。"

    def get_news_info(self, topic: str) -> str:
        """
        获取新闻信息的模拟方法。

        :param topic: 新闻主题。
        :return: 新闻信息字符串。
        """
        # 模拟新闻信息
        return f"关于{topic}的最新新闻：今天发生了一件重要的事情。"

    def display_context(self, task_id: str):
        """
        显示指定任务的上下文信息。

        :param task_id: 任务ID。
        """
        context = self.context_manager.get_context(task_id)
        print(f"\n任务 {task_id} 的上下文：")
        for entry in context:
            role = "用户" if entry['role'] == 'user' else "助手"
            print(f"{role}: {entry['content']}")

# 示例运行
if __name__ == "__main__":
    # 创建上下文管理器
    context_manager = ContextManager(max_length=5)

    # 创建任务处理器
    task_handler = TaskHandler(context_manager)

    # 模拟用户输入并处理任务
    task_handler.handle_task("task1", "北京的天气怎么样？")
    task_handler.handle_task("task1", "今天晚上会下雨吗？")
    task_handler.handle_task("task2", "最近的科技新闻有哪些？")
    task_handler.handle_task("task2", "有没有关于人工智能的报道？")
    task_handler.handle_task("task1", "明天温度多少？")
    task_handler.handle_task("task2", "头条新闻是什么？")
    task_handler.handle_task("task1", "后天还能出太阳吗？")



#例【6-3】
import json
import uuid
import time
from typing import List, Dict, Any, Optional

################################################################################
# MCP 核心数据结构与上下文管理
################################################################################

class Prompt:
    """
    Prompt对象表示一次对话或任务执行中的消息单元。
    """
    def __init__(self, role: str, content: str, parent_id: Optional[str] = None, metadata: Optional[dict] = None):
        self.id = str(uuid.uuid4())
        self.role = role
        self.content = content
        self.parent_id = parent_id
        self.metadata = metadata or {}
        self.timestamp = time.time()

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "role": self.role,
            "content": self.content,
            "parent_id": self.parent_id,
            "metadata": self.metadata,
            "timestamp": self.timestamp
        }

class ContextChain:
    """
    用于管理Prompt链的上下文结构，可以追踪父子Prompt关系。
    """
    def __init__(self):
        self.prompts: Dict[str, Prompt] = {}

    def add_prompt(self, prompt: Prompt):
        self.prompts[prompt.id] = prompt

    def get_prompt_chain(self, end_id: str) -> List[Prompt]:
        """
        根据 end_id 向前追溯上下文链，返回从根到当前节点的Prompt列表。
        """
        chain = []
        current = self.prompts.get(end_id)
        while current:
            chain.insert(0, current)
            if current.parent_id:
                current = self.prompts.get(current.parent_id)
            else:
                break
        return chain

    def print_context(self, end_id: str):
        chain = self.get_prompt_chain(end_id)
        print("=== 上下文链 ===")
        for p in chain:
            print(f"[{p.role}] {p.content} ({p.id[:6]})")

################################################################################
# 可复用任务模块与参数注入
################################################################################

class MailModuleBase:
    """
    邮件任务模块的基类，定义通用的接口和参数注入方法。
    具体模块可继承此类，并实现 process 方法。
    """
    def __init__(self, module_name: str):
        self.module_name = module_name

    def inject_params(self, params: dict):
        """
        可由子类重写，用于将外部参数注入当前模块的内部结构，
        例如将用户输入或上下文信息映射给相应字段。
        """
        raise NotImplementedError

    def process(self) -> str:
        """
        执行模块主要逻辑，并返回字符串结果，供后续Prompt合成或上下文存储。
        """
        raise NotImplementedError


class MailSorter(MailModuleBase):
    """
    邮件整理模块，负责对邮件进行分类，如紧急、普通、广告、系统通知等。
    """
    def __init__(self):
        super().__init__("邮件整理")
        self.mail_content = ""
        self.mail_subject = ""

    def inject_params(self, params: dict):
        self.mail_subject = params.get("subject", "")
        self.mail_content = params.get("content", "")

    def process(self) -> str:
        # 简单逻辑：若主题或内容包含"紧急"则分类为紧急，否则普通
        if "紧急" in self.mail_subject or "紧急" in self.mail_content:
            return f"邮件分类: [紧急]\n主题: {self.mail_subject}"
        else:
            return f"邮件分类: [普通]\n主题: {self.mail_subject}"


class MailFilter(MailModuleBase):
    """
    邮件过滤模块，筛选不需要的邮件，如广告、垃圾邮件等。
    """
    def __init__(self):
        super().__init__("邮件过滤")
        self.classification = ""
        self.mail_content = ""

    def inject_params(self, params: dict):
        self.classification = params.get("classification", "")
        self.mail_content = params.get("content", "")

    def process(self) -> str:
        # 根据分类，若含有敏感词"广告"则过滤，否则通过
        if "广告" in self.mail_content or "广告" in self.classification:
            return "结果: 该邮件被过滤 (广告邮件)"
        return f"结果: 该邮件通过过滤 ({self.classification})"


class MailSender(MailModuleBase):
    """
    邮件发送模块，负责将结果发送给相应收件人。
    """
    def __init__(self):
        super().__init__("邮件发送")
        self.to = ""
        self.body = ""

    def inject_params(self, params: dict):
        self.to = params.get("to", "")
        self.body = params.get("body", "")

    def process(self) -> str:
        return f"邮件已发送给 {self.to}:\n{self.body}"


class MailArchiver(MailModuleBase):
    """
    邮件归档模块，负责将已处理过的邮件存储于归档仓库，供后续查询。
    """
    def __init__(self):
        super().__init__("邮件归档")
        self.archive_path = ""
        self.log_info = ""

    def inject_params(self, params: dict):
        self.archive_path = params.get("archive_path", "/default/archive/")
        self.log_info = params.get("log_info", "")

    def process(self) -> str:
        return f"邮件信息已归档至 {self.archive_path}\n附加日志: {self.log_info}"


################################################################################
# 场景执行：邮件整理 + 邮件过滤 + 邮件发送 + 邮件归档
################################################################################

def run_mail_pipeline():
    """
    演示基于MCP协议思想的可复用任务模块与参数注入流程。
    """

    # 1. 构造上下文管理器
    chain = ContextChain()

    # 2. 添加系统Prompt，描述流程要求
    system_prompt = Prompt(
        role="system",
        content="这是一个邮件处理自动化系统，支持整理、过滤、发送与归档操作。"
    )
    chain.add_prompt(system_prompt)

    # 3. 用户输入，提供邮件主题与内容
    user_prompt = Prompt(
        role="user",
        content="邮件主题: '紧急会议通知'\n邮件正文: '这是一份需要即时处理的紧急会议。'"
    )
    chain.add_prompt(user_prompt)

    # ============ Step A: 邮件整理模块 ============
    sorter = MailSorter()
    sorter.inject_params({
        "subject": "紧急会议通知",
        "content": "这是一份需要即时处理的紧急会议。"
    })
    sort_result = sorter.process()

    # 将处理结果注入上下文
    sorter_prompt = Prompt(
        role="assistant",
        content=sort_result,
        parent_id=user_prompt.id
    )
    chain.add_prompt(sorter_prompt)

    # ============ Step B: 邮件过滤模块 ============
    filter_mod = MailFilter()
    filter_mod.inject_params({
        "classification": "紧急",
        "content": "这是一份需要即时处理的紧急会议。"
    })
    filter_result = filter_mod.process()

    filter_prompt = Prompt(
        role="assistant",
        content=filter_result,
        parent_id=sorter_prompt.id
    )
    chain.add_prompt(filter_prompt)

    # ============ Step C: 邮件发送模块 ============
    # 若通过过滤，才进入发送环节
    if "通过过滤" in filter_result:
        sender = MailSender()
        sender.inject_params({
            "to": "boss@example.com",
            "body": "紧急会议邮件: 需要即时处理，请查看内容。"
        })
        send_result = sender.process()

        send_prompt = Prompt(
            role="assistant",
            content=send_result,
            parent_id=filter_prompt.id
        )
        chain.add_prompt(send_prompt)
    else:
        no_send_prompt = Prompt(
            role="assistant",
            content="该邮件未通过过滤，不会进行发送。",
            parent_id=filter_prompt.id
        )
        chain.add_prompt(no_send_prompt)

    # ============ Step D: 邮件归档模块 ============
    # 进行归档，保存处理记录
    archiver = MailArchiver()
    archiver.inject_params({
        "archive_path": "/mail/archive/2025/urgent/",
        "log_info": f"{sort_result}\n{filter_result}"
    })
    archive_result = archiver.process()
    archive_prompt = Prompt(
        role="assistant",
        content=archive_result,
        parent_id=(send_prompt.id if 'send_prompt' in locals() else filter_prompt.id)
    )
    chain.add_prompt(archive_prompt)

    # 4. 打印最终上下文链，查看执行流程
    chain.print_context(archive_prompt.id)
if __name__ == "__main__":
    run_mail_pipeline()



#例【6-4】
# 定义Prompt Block结构
class PromptBlock:
    def __init__(self, name, template, condition_func, priority=0):
        self.name = name
        self.template = template
        self.condition_func = condition_func
        self.priority = priority
# 示例上下文
context = {
    "user_type": "admin",
    "mail_topic": "系统通知",
    "mail_urgent": True
}
# 构造Prompt Block列表
blocks = [
    PromptBlock(
        name="身份提示",
        template="当前用户为管理员，具有高级处理权限。",
        condition_func=lambda ctx: ctx["user_type"] == "admin",
        priority=1
    ),
    PromptBlock(
        name="紧急提示",
        template="该邮件被标记为紧急，请优先处理。",
        condition_func=lambda ctx: ctx["mail_urgent"],
        priority=2
    ),
    PromptBlock(
        name="普通提示",
        template="邮件主题：{{mail_topic}}，请按照常规流程处理。",
        condition_func=lambda ctx: not ctx["mail_urgent"],
        priority=3
    )
]
# 执行拼接
active_blocks = sorted(
    [b for b in blocks if b.condition_func(context)],
    key=lambda b: b.priority
)
# 填充槽位并拼接
final_prompt = "\n".join(
    b.template.replace("{{mail_topic}}", context["mail_topic"]) for b in active_blocks
)
# 输出拼接结果
print("最终Prompt内容：\n")
print(final_prompt)



#例【6-5】
# 状态定义
class State:
    def __init__(self, name, on_enter):
        self.name = name
        self.on_enter = on_enter  # 状态入口函数

# 状态机类
class StateMachine:
    def __init__(self):
        self.states = {}
        self.current_state = None
    def add_state(self, name, on_enter):
        self.states[name] = State(name, on_enter)
    def set_start(self, name):
        self.current_state = self.states.get(name)
    def transition_to(self, name):
        self.current_state = self.states.get(name)
        if self.current_state:
            self.current_state.on_enter()
# 状态行为函数
def start_state():
    print("[系统] 欢迎使用天气查询助手。请输入查询城市：")
    machine.transition_to("等待输入")
def input_state():
    city = input("用户输入城市: ")
    context["city"] = city
    machine.transition_to("展示结果")
def result_state():
    city = context.get("city", "")
    print(f"[系统] 当前城市 {city} 的天气为：晴，25度，湿度45%。")
# 初始化上下文与状态机
context = {}
machine = StateMachine()
machine.add_state("启动", start_state)
machine.add_state("等待输入", input_state)
machine.add_state("展示结果", result_state)
machine.set_start("启动")
# 执行流程
machine.current_state.on_enter()



#例【6-6】
import threading
import time
# 独立的任务状态容器
class TaskSession:
    def __init__(self, task_id):
        self.task_id = task_id
        self.state = "初始化"
        self.context = []

    def update(self, user_input):
        self.context.append(user_input)
        self.state = f"处理中: {user_input}"

    def finalize(self):
        self.state = "已完成"
# 并发任务处理函数
def run_task(task_id, messages):
    session = TaskSession(task_id)
    for msg in messages:
        session.update(msg)
        print(f"[{task_id}] 当前状态：{session.state}")
        time.sleep(0.5)
    session.finalize()
    print(f"[{task_id}] 最终状态：{session.state}")
# 启动两个并发任务线程
t1 = threading.Thread(target=run_task, args=("任务A", ["查询天气", "查看气温"]))
t2 = threading.Thread(target=run_task, args=("任务B", ["读取邮件", "发送报告"]))
t1.start()
t2.start()
t1.join()
t2.join()